package com.saic.data.dotask;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;

import com.saic.data.main.FTPCheck;
import com.saic.data.table.TableAction;
import com.saic.data.table.TableFile;
import com.saic.data.table.TableUtil;
import com.saic.data.util.CsvParser;
import com.saic.data.util.FTPUtil;
import com.saic.data.util.FileUtil;
import com.saic.data.util.HDFSUtil;
import com.saic.data.util.HbaseUtil;
import com.sun.org.apache.bcel.internal.classfile.PMGClass;

import it.sauronsoftware.ftp4j.FTPFile;

/**
 * @Update item and copy file to HDFS
 * @author Xinshiyou
 */
public class UploadRecord extends DoTask {

	private FTPUtil ftpUtil;
	private HDFSUtil hdfsUtil;
	private FTPFile f;
	private Result res;
	private TableFile table;

	public UploadRecord(FTPUtil ftpUtil, HDFSUtil hdfsUtil, FTPFile f,
			Result res, TableFile table) {
		super();
		this.ftpUtil = ftpUtil;
		this.hdfsUtil = hdfsUtil;
		this.f = f;
		this.res = res;
		this.table = table;
	}

	@Override
	public void doTask() {
		boolean flag;
		String tmpPath;
		try {

			/** begin to parser CSV file */
			tmpPath = "/tmp/"+table.gettConf().getTablename();
			ftpUtil.download(f.getName(),tmpPath, table.gettConf().getFtpDir());
			
			/** copy file to HDFS*/
			/*String localFileName = tmpPath + "/"+ f.getName();
			String hdfsFileName = HDFSUtil.preFix + HDFSUtil.hdfsHome + table.gettConf().getHdfsDir() + "/" + f.getName();
			TableUtil.pm(this.getClass(), "Copy "+localFileName +" to "+hdfsFileName );
			try {
				hdfsUtil.copyTohdfs(localFileName, hdfsFileName);
			} catch (Exception e) {
				// TODO: handle exception
				TableUtil.pm(this.getClass(), "error when copy file to HDFS:" );
				e.printStackTrace();
				return;
			}*/
			
			
			InputStream in = new FileInputStream(new File(tmpPath +"/"+f.getName()));							
			TableUtil.pm(this.getClass(), "null? " + (null == in));		
			List<HashMap<String, String>> lm = CsvParser.Csv2Map_(
					new BufferedReader(new InputStreamReader(in)), false,
					table.gettConf().getToken().charAt(0));
			in.close();
			
			/** begin Insert into csv_tmp */
			int line = 0;
			List<Put> puts = new ArrayList<Put>();
			Put put=null;
			for (HashMap<String, String> map : lm) {
				String rowKey = TableUtil.getKey(table, line);
				map.put("UUID", FileUtil.getUUID());
				map.put("fileKey", new String(res.getRow(),	HbaseUtil.charSet));
				map.put("fileName", table.getFileName());
				map.put("upLoadDate", table.getFinanceDate());
				map.put("streamNo", table.getStreamNo());
				map.put("recordDate", TableUtil.formatDate(new Date()));
				put= HbaseUtil.getPutFromMap(table.gettConf().getTmpHtbl(),table.gettConf().dataFamily,rowKey, map);
				puts.add(put);
				line++;
			}
			HbaseUtil.insertData(table.gettConf().getTmpHtbl(),puts);
			line=puts.size();

			/** Update HBASE FTP-FILE Custom Family */
			Map<String, String> map = TableUtil.getCustomMap(table);
			map.put("hbaseRecordLines", line+"");
			map.put("hbaseRecordTime", String.valueOf(new Date().getTime()));
			put = HbaseUtil.getPutFromMap(TableUtil.configtab,TableUtil.configCustomFamily,
					table.getRowkey(), map);
			flag =HbaseUtil.insertData(TableUtil.configtab,put);
			TableUtil.pm(this.getClass(), "Update hbase ftp_file:CustomFields "
					+ (flag ? "SUCCESS" : "FAILER"));

			/** Update HBASE FTP-FILE fileStats */
			flag = HbaseUtil.updateData(TableUtil.configtab, res.getRow(),
					TableUtil.configtabFamily,"fileStats", TableAction.UpdateDB.toString()+"");
			TableUtil.pm(this.getClass(), "Update hbase file Stats "
					+ (flag ? "SUCCESS" : "FAILER"));
			
			/** Delete from FTP */
			if (FTPCheck.deleteflag){
			flag = ftpUtil.delete(f);
			TableUtil.pm(this.getClass(), "Delete " + f.getName() + " from FTP "
					+ (flag ? "SUCCESS" : "FAILER"));}

		} catch (IllegalArgumentException e) {
			TableUtil.pm(this.getClass(), "IllegalArgumentException:"+e.getMessage());
			e.printStackTrace();
		} catch (IOException e) {
			TableUtil.pm(this.getClass(), "IOException:"+e.getMessage());
			e.printStackTrace();
		} catch (Exception e) {
			TableUtil.pm(this.getClass(), e.getMessage());
			e.printStackTrace();
		}
	}

	@Override
	public Object getResult() {
		return result;
	}

}
